CxfInputFormat.java
 
 package com.alibaba.cxf.sort;
 
 import java.io.IOException;
 
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.FileInputFormat;
 import org.apache.hadoop.mapred.FileSplit;
 import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.LineRecordReader;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
 
 public class CxfInputFormat extends FileInputFormat<IntWritable,Text>{
  @Override
  public RecordReader<IntWritable, Text> getRecordReader(InputSplit split,
    JobConf job, Reporter reporter) throws IOException {
   return new CxfRecordReader(job, (FileSplit) split);
  }
  class CxfRecordReader implements RecordReader<IntWritable,Text> {
 
   private LineRecordReader in;
      private LongWritable junk = new LongWritable();
      private Text line = new Text();
      private  int KEY_LENGTH = 10;
   public CxfRecordReader(JobConf job,FileSplit split) throws IOException{
    in = new LineRecordReader(job, split);
   }
   @Override
   public void close() throws IOException {
    in.close();   
   }
   @Override
   public IntWritable createKey() {
    return new IntWritable();
   }
   @Override
   public Text createValue() {
    
    return new Text();
   }
   @Override
   public long getPos() throws IOException {
    
    return in.getPos();
   }
   @Override
   public float getProgress() throws IOException {
    
    return in.getProgress();
   }
   @Override
   public boolean next(IntWritable key, Text value) throws IOException {
    if (in.next(junk, line)) {
     if (line.getLength() < KEY_LENGTH) {
      key.set(Integer.parseInt(line.toString()));
      value = new Text();
   //   value.clear();
     } else {
      byte[] bytes = line.getBytes();
      key.set(Integer.parseInt(new String(bytes).substring(0, KEY_LENGTH)));
      value = new Text();
     }
     return true;
    } else {
     return false;
    }
   }
  }
 }
 
  
 
 SortByMapReduce.java
 
 package com.alibaba.cxf.sort;
 
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import org.apache.hadoop.filecache.DistributedCache;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapred.FileInputFormat;
 import org.apache.hadoop.mapred.FileOutputFormat;
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.TextOutputFormat;
 import org.apache.hadoop.mapred.lib.InputSampler;
 import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
 public class SortByMapReduce {
 
  /**
   * @param args
   * @throws URISyntaxException
   * @throws IOException
 */
  public static void main(String[] args) throws IOException, URISyntaxException {
   runJob(args);
  }
 
  private static void runJob(String[] args) throws IOException, URISyntaxException {
   
   JobConf conf = new JobConf(SortByMapReduce.class);
   
   FileInputFormat.setInputPaths(conf, new Path(args[0]));
         FileOutputFormat.setOutputPath(conf, new Path(args[1]));
         conf.setJobName(”SortByMapReduce”);
   
   conf.setInputFormat(CxfInputFormat.class);
   conf.setOutputKeyClass(IntWritable.class);
   conf.setOutputFormat(TextOutputFormat.class);
   conf.setNumReduceTasks(5);
   conf.setPartitionerClass(TotalOrderPartitioner.class);
   InputSampler.RandomSampler<IntWritable, NullWritable> sampler =
    new InputSampler.RandomSampler<IntWritable, NullWritable>(0.1,10000,10);
   
   Path input = FileInputFormat.getInputPaths(conf)[0];
   input = input.makeQualified(input.getFileSystem(conf));
   Path partitionFile = new Path(input,”_partitions”);
   TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
   InputSampler.writePartitionFile(conf, sampler);
   
   URI partitionURI = new URI(partitionFile.toString() + “#_partitions”);
   DistributedCache.addCacheFile(partitionURI, conf);
   DistributedCache.createSymlink(conf);
   JobClient.runJob(conf);  
  } 
 }
